package wili.demo.misc;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

/**
 * 
 * @author weila 2020年1月26日
 */
class Stream_parallel_demo {
	public class ForkJoinCalculate extends RecursiveTask<Long> {
		private static final long serialVersionUID = 13475679780L;

		private long start;

		private long end;

		private static final long THRESHOLD = 10000L; //临界值

		public ForkJoinCalculate(long start, long end) {
			this.start = start;
			this.end = end;
		}

		@Override
		protected Long compute() {
			long length = end - start;
			if (length <= THRESHOLD) {
				long sum = 0;
				for (long i = start; i <= end; i++) {
					sum += i;
				}
				return sum;
			} else {
				long middle = (start + end) / 2;
				ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
				left.fork(); //拆分，并将该子任务压入线程队列			
				ForkJoinCalculate right = new ForkJoinCalculate(middle + 1, end);
				right.fork();
				return left.join() + right.join();
			}
		}
	}

	public void test1() throws InterruptedException, ExecutionException {
		long start = System.currentTimeMillis();
		ForkJoinTask<Long> task = new ForkJoinCalculate(0L, max);
		long sum;
		if ("safd".isEmpty()) {
			sum = task.invoke();
		} else {
			ForkJoinPool pool = new ForkJoinPool();
			sum = pool.invoke(task);
		}
		long end = System.currentTimeMillis();
		System.out.println(sum + ", 耗费的时间为: " + (end - start)); //112-1953-1988-2654-2647-20663-113808
	}

	static final long max = (long) 1e9;// 1e10;

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		//copy from https://blog.csdn.net/rubulai/article/details/89022315
		/*
		fork/join 框架的优势体现在对其中包含的任务的处理方式上。在一般的线程池中，
		如果一个线程正在执行的任务由于某些原因无法继续运行，那么该线程会处于等待状态，
		而在 fork/join 框架实现中，如果某个子任务由于等待另外一个子任务的完成而无法继续运行。
		那么处理该子任务的线程会主动寻找其他尚未运行的子任务来执行，这种方式减少了线程的等待时间，提高了性能。
		
		 */
		int opt = 1;
		opt = 2;
		if (opt == 1) {
			//Fork/Join框架在JDK1.7是就有了，但是使用起来相对繁琐,如下：
			Stream_parallel_demo demo = new Stream_parallel_demo();
			demo.test1();
		} else if (opt == 2) {
			Instant start, end;
			long ret;
			if ("sdaf".isEmpty()) {
				/*
					使用Java8的并行流：底层依然是Fork/Join框架.并行流时默认会根据机器的CPU核心数开启线程，
					本例中，数据量超过亿时，速度才能快1倍以上。
				 */
				start = Instant.now();
				ret = LongStream.rangeClosed(0, max)// 获取一个0-1千亿的元素的流
						.parallel()// 转为并行流
						//.sequential()//串行流，更慢？
						.reduce(0, Long::sum);// 累加求和
				end = Instant.now();
			} else {
				start = Instant.now();
				ret = 0;
				for (long i = 0; i <= max; i++) {
					ret += i;
				}
				end = Instant.now();
			}
			System.out.println(ret + ", " + Duration.between(start, end).toMillis());
		} else if (opt == 3) {
		}
	}
}
